package com.whsxt.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.rabbitmq.client.Channel;
import com.whsxt.constant.QueueConstant;
import com.whsxt.dao.ProdEsDao;
import com.whsxt.domain.Prod;
import com.whsxt.es.ProdEs;
import com.whsxt.service.ImportService;
import com.whsxt.service.ProdService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @Author: caoqingwen
 * @CreateTime: 2021-03-17 10:57
 * @Description:
 */
@Service
@Slf4j
public class ImportServiceImpl implements ImportService, CommandLineRunner {


    @Autowired
    private ProdEsDao prodEsDao;

    @Autowired
    private ProdService prodService;

    @Value("${esimport.size}")
    private Integer size;

    /**
     * 初始时间
     */
    private Date t1;

    /**
     * 全量导入
     */
    @Override
    public void importAll() {
        log.info("开始全量导入====================================");
        //计算分页
        //1、查询所有的count
        Integer totalCount = prodService.getTotalCount(null, null);
        if (totalCount == null || totalCount == 0) {
            log.info("没有需要导入的数据");
            return;
        }
        //2、计算分页
        long page = totalCount % size == 0 ? totalCount / size : (totalCount / size) + 1;
        //3、循环页数
        for (int i = 1; i <= page; i++) {
            //查询导入
            importToEs(i, size, null, null);
        }
        log.info("全量导入完成======================================");

        //记录一个时间，作为增量导入的初始时间
        t1 = new Date();
    }


    /**
     * 导入es的方法
     *
     * @param current
     * @param size
     * @param t1
     * @param t2
     */
    private void importToEs(int current, Integer size, Date t1, Date t2) {
        //1、分页查询
        //组装一个page
        Page<Prod> page = new Page<>(current, size, false);
        Page<Prod> prodPage = prodService.findProdByPageToEs(page, t1, t2);
        List<Prod> prodList = prodPage.getRecords();
        //2、组装数据
        List<ProdEs> prodEsList = new ArrayList<>();
        prodList.forEach(prod -> {
            ProdEs prodEs = new ProdEs();
            //对象拷贝
            BeanUtil.copyProperties(prod, prodEs, true);
            prodEsList.add(prodEs);
        });
        //3、导入
        prodEsDao.saveAll(prodEsList);
    }

    /**
     * 增量导入
     * initialDelay:项目启动隔**长时间执行
     * fixedRate:第一次执行后，每隔多长时间执行
     */
    @Override
    @Scheduled(initialDelay = 3 * 60 * 1000, fixedRate = 3 * 60 * 1000)
    public void importUpdate() {
        log.info("增量导入开始===========================================");
        //t2比t1大30分钟
        Date t2 = new Date();

        //计算分页
        //1、查询所有的count
        Integer totalCount = prodService.getTotalCount(t1, t2);
        if (totalCount == null || totalCount == 0) {
            log.info("没有需要导入的数据");
            //更新t1
            t1 = t2;
            return;
        }
        //2、计算分页
        long page = totalCount % size == 0 ? totalCount / size : (totalCount / size) + 1;
        //3、循环页数
        for (int i = 1; i <= page; i++) {
            //查询导入
            importToEs(i, size, t1, t2);
        }
        log.info("增量导入完成======================================");

        //更新t1
        t1 = t2;
    }


    @Override
    public void quickImport() {


    }


    /**
     * 快速导入:用户下单减库存，mq实现
     * 用户下单   order-service    我们做一个解耦合
     * 用户下单量特别大，如果使用openfeign对服务的压力很大，而且性能不好
     * 使用mq 特点：异步，削峰，限流，解耦合
     * 业务流程：
     * 1、用户下订单，会修改mysql库存，需要同步ES
     * 2、把减库存的消息放到mq，   prodId(商品id，spu)   count(订单数量)      数据结构 Map<Long,Integer>
     * 3、处理消息
     */
    @RabbitListener(queues = {QueueConstant.PROD_CHANGE_QUEUE}, concurrency = "3-5")
    public void quickImport(Message message, Channel channel) {
        //拿到消息体
        String msgStr = new String(message.getBody());
        //把数据转成map集合
        JSONObject prodInfo = JSON.parseObject(msgStr);
        //消费消息
        List<Long> prodIds = new ArrayList<>();
        prodInfo.forEach((prodId, count) -> {
            //根据id，查询es，查到要修改的对象
            prodIds.add(Long.valueOf(prodId));
        });
        //查询es
        Iterable<ProdEs> prodChangeList = prodEsDao.findAllById(prodIds);
        //循环 修改数据
        prodChangeList.forEach(prodEs -> {
            Long prodId = prodEs.getProdId();
            //改变的数量
            Integer count = prodInfo.getInteger(String.valueOf(prodId));
            prodEs.setTotalStocks(prodEs.getTotalStocks() + count);
        });

        //插入es
        prodEsDao.saveAll(prodChangeList);
        //签收消息
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            channel.basicAck(deliveryTag, false);
            log.info("同步ES成功");
        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    /**
     * 不加注解实现项目启动就执行，可以使用实现CommandLineRunner
     * 当IOC容易初始化完成，就会执行这个方法
     *
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        importAll();
    }


}
